package org.example.kafka;

import com.alibaba.fastjson.JSONObject;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.example.websocket.common.WebSocketServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController;

import java.util.Optional;

/**
 * @version : V1.0
 * @Description : 消费者
 */
@Component
@Slf4j
@RestController
@Order
@Data
public class KafkaConsumer {

    @Value("${logging.log-print-switch}")
    private String logprintswitch;

    @Autowired
    private WebSocketServer webSocketServer;

    /**
     * 读取kafka数据
     * @param record
     * @param topic->signaler_rt_phase
     */
    @KafkaListener(topics = "test_topic", groupId = "test_group")
    public void topic_message(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            JSONObject jsonObject = JSONObject.parseObject(msg.toString());
            //打印log日志
            if ("1".equals(logprintswitch)) {
                log.info( "Topic:" + topic + ",Message:" + msg);
            }
            try {
                webSocketServer.onMessage(message.toString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
